package com.lxl.hadoop.flowtopn;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Map;
import java.util.TreeMap;

/**
 * @Author lixilin
 * @Date 2022/1/12 20:58
 * @Description
 */
public class TopNReducer extends Reducer<TopNFlowBean, Text,Text,Text> {

    TreeMap<TopNFlowBean,Text> treeMap = new TreeMap();
    Text k = new Text();
    Text v = new Text();
    StringBuilder sb = new StringBuilder();

    @Override
    protected void reduce(TopNFlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Text phone = new Text();
        TopNFlowBean flowBean = new TopNFlowBean();
        flowBean.set(key.getUpFlow(),key.getDownFlow());
        for (Text value : values) {
            phone.set(value);
            treeMap.put(flowBean, phone);
            if (treeMap.size() >10) {
                treeMap.pollLastEntry();
            }
        }
    }
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {

        for (Map.Entry<TopNFlowBean, Text> entry : treeMap.entrySet()) {
            sb.setLength(0);
            TopNFlowBean bean = entry.getKey();
            k.set(entry.getValue());
            sb.append(bean.getUpFlow());
            sb.append("\t");
            sb.append(bean.getDownFlow());
            sb.append("\t");
            sb.append(bean.getSumFlow());
            v.set(sb.toString());
            context.write(k,v);


        }
    }
}
